package com.sunlands.deskmate.message.handler.sender.trigger;

import com.google.common.util.concurrent.ThreadFactoryBuilder;
import com.sunlands.deskmate.config.AsyncExecutorConfig;
import com.sunlands.deskmate.message.entity.Messages;
import com.sunlands.deskmate.channel.ChannelInfo;
import com.sunlands.deskmate.message.handler.sender.Sender;
import com.sunlands.deskmate.message.handler.sender.websocket.DefaultWebsocketMessageSender;
import com.sunlands.deskmate.message.handler.storager.unsync.UnsyncMessageStorager;
import lombok.extern.slf4j.Slf4j;
import org.redisson.api.RLock;
import org.redisson.api.RedissonClient;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
import javax.annotation.PostConstruct;
import java.util.Objects;
import java.util.concurrent.*;
import static com.sunlands.deskmate.message.handler.mq.StorageMessageWorker.leaseTimeInSeconds;
import static com.sunlands.deskmate.message.handler.mq.StorageMessageWorker.waitTimeInSeconds;


/**
 * 消息发送触发器，触发器由事件驱动，事件来自消息接受
 * @author anjunling@sunlands.com
 */
@Slf4j
@Component
public class SendMessageTrigger {

    @Autowired
    private RedissonClient redissonClient;

    @Autowired
    private UnsyncMessageStorager storager;

    @Autowired
    UnsyncMessageAcquire unsyncMessageAcquire;

    private BlockingQueue<TriggerEvent> queue = new LinkedBlockingQueue<>();

    private static ThreadFactory sendMessageThreadFactory = new ThreadFactoryBuilder().setNameFormat("send-user-message-pool-%d").build();

    private static final ExecutorService sendMessagePool = new ThreadPoolExecutor(1, 4, 1, TimeUnit.HOURS, new LinkedBlockingDeque<>(1024), sendMessageThreadFactory, new AsyncExecutorConfig.AbortPolicy());
    private boolean isDone;

    @PostConstruct
    public void dispatch() {
        Worker worker = new Worker(this);
        Thread thread = new Thread(worker);
        thread.start();
    }

    public boolean offer(TriggerEvent dispatchEvent) {
        try {
            return queue.offer(dispatchEvent, 5, TimeUnit.SECONDS);
        } catch (InterruptedException e) {
            log.error("queue size " + queue.size(), e);
            Thread.currentThread().interrupt();
            return false;
        }
    }

    //TODO 这个和offer的区别?
    public boolean epochNext(TriggerEvent dispatchEvent) {
        boolean success;

        //TODO ??
        unlockEpoch(dispatchEvent);

        long size = storager.countUnreadMessage(dispatchEvent.getSessionKey());

        if (size > 0) {
            success = offer(dispatchEvent);

            log.debug("offer dispatchEvent={}, result={}", dispatchEvent, success);
        } else {
            //TODO 为什么没有size就不发送?
            success = false;
            log.debug("no unread message, sessionKey={}", dispatchEvent.getSessionKey());
        }

        return success;
    }

    public void unlockEpoch(TriggerEvent dispatchEvent) {
        String key = lockKey(dispatchEvent);
        RLock lock = redissonClient.getLock(key);
        if(Objects.nonNull(lock)){
            try {
                lock.forceUnlock();
            } catch (Exception ignored) {
                log.error("unlock error ", ignored);
            }
        }
        log.info("unlock event key={}", key);
    }

    public boolean lockEpoch(TriggerEvent dispatchEvent) {
        String key = lockKey(dispatchEvent);
        RLock lock = redissonClient.getLock(key);
        boolean result = false;

        if(Objects.nonNull(lock) && lock.isLocked()){
            return false;
        }

        try {
            result = lock.tryLock(waitTimeInSeconds, leaseTimeInSeconds, TimeUnit.SECONDS);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        log.info("lockEpoch key={}, result={}", key, result);
        return result;
    }

    private String lockKey(TriggerEvent dispatchEvent) {
        return dispatchEvent.getEvent() + "_" + dispatchEvent.getSessionKey();
    }

    public void release() {
        isDone = true;
    }

    /**
     *
     */
    private class Worker implements Runnable {

        private SendMessageTrigger sendDispatcher;

        public Worker(SendMessageTrigger sendDispatcher) {
            this.sendDispatcher = sendDispatcher;
        }

        @Override
        public void run() {
            while (!sendDispatcher.isDone || Thread.currentThread().isInterrupted()) {
                TriggerEvent dispatchEvent = null;
                try {
                    dispatchEvent = take();
                    log.debug("take dispatchEvent={}", dispatchEvent);
                    CompletableFuture.completedFuture(dispatchEvent)
                            .thenAcceptAsync(event -> {
                                boolean success = lockEpoch(event);
                                Sender sender;
                                if (success) {
                                    log.debug("dispatcher, lockEpoch result={}, event={}, msgId={}", success, event, event.getMsgId());
                                    if ((sender = getSender(event)) == null) {
                                        unlockEpoch(event);
                                        log.info("sender not found, event={}", event);
                                    }else {
                                        sender.send();
                                    }
                                }else{
                                    log.info("未获取到锁，抛掉消息 dispatchEvent = {}", event);
                                }
                            }, sendMessagePool);
                } catch (InterruptedException ie) {
                    log.error("dispatch interrupted exception");
                    Thread.currentThread().interrupt();
                    //重新启动
                    dispatch();
                } catch (Exception e) {
                    log.error("dispatch event " + dispatchEvent, e);
                }
            }
        }

        private Sender getSender(TriggerEvent dispatchEvent) {
            if (dispatchEvent == null) {
                log.warn("take is empty");
                return null;
            }

            //判断是否消息已同步
            //FIXME msgId可能为空
            Long eventMsgId = dispatchEvent.getMsgId();
            Long msgIdLong = storager.getSyncFlagValue(dispatchEvent.getSessionKey());

            if (Objects.nonNull(eventMsgId) && Objects.nonNull(msgIdLong) && eventMsgId != 0 && msgIdLong >= eventMsgId) {
                log.debug("syncId={} > event msgId={}, skip event", msgIdLong, eventMsgId);
                return null;
            }

            Messages messages = unsyncMessageAcquire.getUnreadMessage(dispatchEvent.getSessionKey());
            //获取待发送消息
            if (messages == null) {
                log.debug("unread message is empty, sessionKey={}", dispatchEvent.getSessionKey());
                return null;
            }

            Sender sender = findSender(dispatchEvent, messages);
//            if (sender == null) {
//                log.warn("dispatch event {} sender not exists", dispatchEvent);
//                return null;
//            }

            return sender;
        }

        private TriggerEvent take() throws InterruptedException {
            TriggerEvent dispatchEvent;
            try {
                dispatchEvent = queue.take();
            } catch (InterruptedException e) {
                log.warn("take retry");
                throw e;
            }
            return dispatchEvent;
        }

        private Sender findSender(TriggerEvent dispatchEvent, Messages messages) {
//            ClientInfoManager.ClientInfo clientInfo = clientInfoManager
//                    .clientInfo(dispatchEvent.getClientId(), dispatchEvent.getUserIdentity());
//
//            if (clientInfo == null) {
//                log.info("clientInfo not found, dispatchEvent={}", dispatchEvent);
//                return null;
//            }
            String sessionKey = dispatchEvent.getSessionKey();
            ChannelInfo channelInfo = ChannelInfo.parseSessionKey(sessionKey);
//            if (channelInfo == null) {
//                return null;
//            }

//            try {
                return new DefaultWebsocketMessageSender(channelInfo, messages);
//                if (ProtocolEnum.HTTP.getType().equalsIgnoreCase(clientInfo.getCallbackProtocol())) {
//                    //HTTP Sender
//                    convert(clientInfo, channelInfo, messages);
//                    return new DefaultHttpMessageSender(channelInfo, clientInfo.getCallbackUrl(), messages);
//
//                } else if (ProtocolEnum.HTTPS.getType().equalsIgnoreCase(clientInfo.getCallbackProtocol())) {
//                    //HTTPS Sender
//
//                } else if (ProtocolEnum.WEBSOCKET.getType().equalsIgnoreCase(clientInfo.getCallbackProtocol())) {
//                    //WS Sender
//                    convert(clientInfo, channelInfo, messages);
//                } else if (ProtocolEnum.WEBSOCKETS.getType().equalsIgnoreCase(clientInfo.getCallbackProtocol())) {
//                    //WSS Sender
//
//                }
//            } catch (Exception e) {
//                log.error("exception in findSender", e);
//            }

//            return null;
        }

        /**
         * 获取对接方用户ID
         *
         * @param clientInfo
         * @param channelInfo
         * @param messages
         * @throws Exception
         */
//        private void convert(ClientInfoManager.ClientInfo clientInfo, ChannelInfo channelInfo, List<OutboundMessage> messages) {
//            if (IdentityEnum.USER.identity.equalsIgnoreCase(clientInfo.getUserIdentity())) {
//
//                for (OutboundMessage message : messages) {
//                    String openId = userManager.convert(clientInfo, message.getToId(), message.getAppId());
//                    if (StringUtils.isEmpty(openId)) {
//                        //删除无效用户数据
//                        String clientIdentityUserKey = channelInfo.getSessionKey();
//                        CompletableFuture.completedFuture(clientIdentityUserKey).thenAcceptAsync(id -> storager.del(id));
//
//                        throw new IllegalStateException("userId " + channelInfo.getUserId() + " convert thirdId is null");
//                    }
//                    message.setToId(openId);
//                }
//            }
//        }

    }



}
